package com.companyname.projecttemplate.component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import com.base.core.context.utils.LoggerWrapper;
import com.companyname.projecttemplate.ws.AppWebSocketHandler;

import lombok.extern.slf4j.Slf4j;

/**
 * 发送消息
 * @author start
 */
@Slf4j
@Service
public class SendService {
	
	@Autowired
	private RocketMQTemplate rocketmqTemplate;

	/**
	 * mq消息
	 * @param <T>
	 * @param topic
	 * @param ao
	 */
	public <T> void mq(String topic, T ao) {
		rocketmqTemplate.asyncSend(topic, ao, new SendCallback() {

			@Override
			public void onSuccess(SendResult sendResult) {
				if(log.isInfoEnabled()) {
					log.info("mq push sendResult:{}", sendResult);
				}
			}

			@Override
			public void onException(Throwable e) {
				LoggerWrapper.print(e, false);
			}

		});
	}
	
	/**
	 * ws消息
	 * @param userId
	 * @param wm
	 * @param isBinary
	 */
	 public static void ws(String userId,String content,Boolean isBinary) {
    	if(!AppWebSocketHandler.SESSIONPOOL.containsKey(userId)) {
    		//消息被丢弃
    		return;
    	}
    	WebSocketSession session=AppWebSocketHandler.SESSIONPOOL.get(userId);
    	synchronized (session) {
    		if(session.isOpen()){
                try {
                	if(isBinary) {
                		session.sendMessage(new BinaryMessage(content.getBytes(StandardCharsets.UTF_8)));
                	}else {
                		session.sendMessage(new TextMessage(content));
                	}
                }catch (IOException e){
        			log.error("消息下发异常", e);
                }
            }
		}
    }

}
